在前面的文章中,我們從非同步概念、協程、事件迴圈,一路探索到執行緒、程序以及 GIL。現在,讓我們用這些知識來回答一個所有 FastAPI 開發者都會遇到的核心問題:
在 FastAPI 中,def
與 async def
定義的 API 路由,到底有什麼差別?我該在什麼時候用哪一個?
要做出最好的選擇,我們必須理解 FastAPI 在背後做了什麼。接下來讓我們串連起之前學到的所有知識。
當你使用 async def
定義路由時,你的程式碼將直接在 FastAPI 底層的事件迴圈 (Event Loop) 上執行(由 Uvicorn 等 ASGI 伺服器提供)。這種執行方式的精髓在於 await
關鍵字:當事件迴圈執行到 await
一個 I/O 操作時(例如 await db.query()
),它不會原地等待,而是會「掛起」當前任務,立即轉去處理其他進來的請求。當原本的 I/O 操作完成後,再回來繼續執行剩餘的程式碼。
這種機制讓單一執行緒就能實現極高的並行處理能力 (Concurrency),特別適合處理大量的網路請求。
程式碼範例:
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# 使用非同步資料庫驅動 (如 databases, asyncpg)
user_data = await db.users.get(id=user_id)
# 使用非同步 HTTP 客戶端 (如 httpx)
external_data = await httpx.get(f"https://api.example.com/data/{user_id}")
return {"user": user_data, "external": external_data.json()}
相對地,當你使用 def
定義路由時,FastAPI 會採取完全不同的處理策略。由於 FastAPI 知道這個函式可能會包含阻塞操作,為了保護珍貴的事件迴圈不被卡住,它不會讓這些函式直接在主事件迴圈上執行。取而代之的是,FastAPI 會從一個外部執行緒池 (External Thread Pool) 中分配一個執行緒,然後將你的 def
函式放到那個獨立的執行緒中執行。
這樣一來,即使函式內部有 time.sleep(10)
或複雜的 CPU 運算,也只會阻塞那個被分配到的執行緒,而主事件迴圈依然能夠暢通無阻地接收和處理其他請求(特別是 async def
的請求)。
程式碼範例:
def complex_cpu_calculation(data: dict) -> int:
# 這裡是一個耗時的 CPU 運算
# ...
time.sleep(10) # 模擬阻塞操作
return sum(data.values())
@app.post("/calculate")
def run_calculation(data: dict):
result = complex_cpu_calculation(data)
return {"result": result}
既然 def
會在執行緒池中執行,這是否表示我們可以透過 def
路由來實現 CPU 密集型任務的並列 (Parallelism),從而利用多核心 CPU 呢?
答案:不行,因為有 GIL。
GIL 會確保同一個程序中,任何時候都只有一個執行緒在執行 Python 位元組碼。即使你的伺服器有 16 個核心,多個執行緒的 def
路由也無法「真正地同時」進行 CPU 運算。它們會爭搶 GIL,導致頻繁的上下文切換,效能甚至可能比單執行緒更差。
現在,你可以根據以下簡單的原則來做決定:
使用 async def
:
async def
讓 FastAPI 能在「等待」I/O 時處理其他請求,最大化伺服器吞吐量使用 def
:
asyncio
的傳統阻塞函式庫時def
路由放到一個獨立的執行緒池中執行,避免阻塞主事件迴圈async def
中呼叫阻塞函式!這是最常見也最嚴重的錯誤:
import requests
@app.get("/bad-request")
async def bad_request():
# requests.get() 是一個阻塞函式
# 它會凍結整個事件迴圈,所有其他請求都會被卡住!
response = requests.get("https://example.com")
return response.json()
修正方法:要嘛改用 def
,要嘛換用非同步函式庫(如 httpx)。
如果一個 async def
函式必須呼叫一個無法避免的阻塞函式,可以使用 fastapi.concurrency.run_in_threadpool
:
from fastapi.concurrency import run_in_threadpool
def blocking_io_call():
# ...
pass
@app.get("/mixed")
async def handle_mixed():
# 將阻塞函式放到執行緒池,然後 await 它的結果
# 這能避免阻塞事件迴圈
result = await run_in_threadpool(blocking_io_call)
return {"status": "ok"}
有關 thread pool 的部分,會在後面的文章做更完整的介紹。
def
和 async def
在 FastAPI 中並非對立,而是相輔相成的工具,它們共同構成了框架的彈性與強大。
async def
是 FastAPI 的「預設」與「未來」:它利用事件迴圈處理 I/O 密集型任務,以實現最大的伺服器效能。def
是強大的「兼容模式」:它利用執行緒池確保了 CPU 密集型任務和傳統阻塞函式庫不會拖垮整個應用。理解了它們背後的非同步、執行緒與 GIL 原理,你就能夠針對每個具體場景,設計出最高效、最穩定的 FastAPI 應用程式。